package org.lql.chaos.kafka.example.interceptor;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.ProducerInterceptor;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

import java.util.Map;

/**
 * @author: liangqinglong
 * @date: 2025-08-10 01:35
 * @description:
 **/
@Slf4j
public class ExampleProducerInterceptor implements ProducerInterceptor<String, Object> {
	@Override
	public ProducerRecord<String, Object> onSend(ProducerRecord<String, Object> record) {
		return new ProducerRecord<>(
				record.topic(),
				record.partition(),
				record.timestamp(),
				record.key(),
				record.value(),
				record.headers());
	}

	@Override
	public void onAcknowledgement(RecordMetadata recordMetadata, Exception exception) {
		if (exception != null) {
			log.error("发送失败: {}", exception.getMessage());
		}
	}

	@Override
	public void close() {
		log.info("关闭拦截器");
	}

	@Override
	public void configure(Map<String, ?> map) {

	}
}
